package com.wujilee.wujileenetty.demos.web.netty.heima;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;


/**
 * @author liwuji
 * @date 2024/8/6
 * @desc
 **/
@Data
@Slf4j
public class Worker implements Runnable {
    private Thread thread;
    private String name;
    private Selector selector;
    private volatile boolean start = false;
    private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();
    private AtomicBoolean isRegister =  new AtomicBoolean(false);

    public void register(SocketChannel socketChannel)  {
        if (!start) {
            thread = new Thread(this, this.name);
            try {
                this.selector = Selector.open();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            start = true;
            thread.start();
        }

//        queue.add(() -> {
//            try {
//                socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
//            } catch (ClosedChannelException e) {
//                throw new RuntimeException(e);
//            }
//            try {
//                Thread.sleep(10L);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
          if(isRegister.compareAndSet(false, true)){
              selector.wakeup();
              try {
                  socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
              } catch (ClosedChannelException e) {
                  throw new RuntimeException(e);
              }
              isRegister.compareAndSet(true, false);
          }
    }

    @Override
    public void run() {
        try {
            while (true) {
                if(!isRegister.compareAndSet(false, true)){
                    Thread.sleep(1L);
                }
                selector.select();
//                Runnable task = queue.poll();
//                if (task != null) {
//                    task.run();
//                }
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()){
                    SelectionKey selectionKey = iterator.next();
                    try {
                        if (selectionKey.isReadable()) {
                            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                            ByteBuffer attachment = (ByteBuffer) selectionKey.attachment();
                            int read = socketChannel.read(attachment);
                            if(read == -1){
                                selectionKey.cancel();
                                log.info("客户端断开连接");
                            }
                            attachment.flip();
                            byte[] data = new byte[attachment.limit()];
                            attachment.get(data);
                            log.info("服务端接收到数据："+new String(data));
                            attachment.clear();
                        }
                    } catch (Exception e) {
                        throw new RuntimeException("读取异常", e);
                    }finally {
                        iterator.remove();
                    }
                }

            }
        } catch (Exception e) {
            throw new RuntimeException("执行异常", e);
        }

    }
}
